package com.pig4cloud.pigx.contract.api.v1.websocket.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.common.customize.api.enums.AbnormalEnum;
import com.pig4cloud.pigx.common.customize.api.exception.OpenApiException;
import com.pig4cloud.pigx.common.customize.i18n.util.MessageUtils;
import com.pig4cloud.pigx.contract.api.v1.utils.OpenApiContractUtilsV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.dto.OpenApiKLineDTOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.message.OpenApiSendMessageV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.vo.OpenApiAbstractTopicVOV1;
import com.pig4cloud.pigx.contract.constant.ContractKLineConstants;
import com.pig4cloud.pigx.contract.entity.ContractKLine;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class OpenApiKlineV1 extends RequestDataAbstractTopicV1 {

	private final OpenApiContractUtilsV1 apiContractUtils;

	private final ContractUtils contractUtils;

	public static final String API_NAME = "k_line";

	public OpenApiKlineV1(OpenApiContractUtilsV1 apiContractUtils, ContractUtils contractUtils) {
		this.apiContractUtils = apiContractUtils;
		this.contractUtils = contractUtils;
		init();
	}

	private void init() {
		wsPush();
	}

	private void wsPush() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			Map<String, List<WebSocketSession>> map = new HashMap<>();
			// 记录请求参数不合法的session,发送一次报错信息后,移除session队列
			List<WebSocketSession> requestErrorList = new ArrayList<>();
			// 根据参数对session分组
			for (Map.Entry<WebSocketSession, OpenApiAbstractTopicVOV1> entry : subList.entrySet()) {
				if (!entry.getKey().isOpen()) {
					subList.remove(entry.getKey());
					log.info("ws 移除已关闭的session");
					continue;
				}
				try {
					// 开始参数校验
					JSONObject param = entry.getValue().getData().getRequestParam();
					OpenApiKLineDTOV1 dto = JSONObject.parseObject(param.toJSONString(), OpenApiKLineDTOV1.class);
					if (dto == null
							|| dto.getSymbolName() == null
							|| dto.getType() == null
							|| dto.getSymbolName().isEmpty()
							|| dto.getType().isEmpty()
					) {
						// 必填参数为空发送异常消息
						entry.getKey().sendMessage(OpenApiSendMessageV1.getTextMessage(R.failed(
								AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getCode(),
								MessageUtils.get(AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getMessage()))
						));
						requestErrorList.add(entry.getKey());
						continue;
					}
					// 错误的type类型
					if (ContractKLineConstants.findByTimeRangeType(dto.getType()) == null) {
						entry.getKey().sendMessage(OpenApiSendMessageV1.getTextMessage(R.failed(
								AbnormalEnum.MSG.UNKNOWN_ORDER_COMPOSITION.getCode(),
								MessageUtils.get(AbnormalEnum.MSG.UNKNOWN_ORDER_COMPOSITION.getMessage(),"type"))
						));
						requestErrorList.add(entry.getKey());
						continue;
					}

					// 以币对名称 + k线类型 为key对session分组
					String key = dto.getSymbolName() + "," + dto.getType();
					if (map.containsKey(key)) {
						map.get(key).add(entry.getKey());
					} else {
						List<WebSocketSession> sessionList = new ArrayList<>();
						sessionList.add(entry.getKey());
						map.put(key, sessionList);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			// 移除请求参数异常的session
			for (WebSocketSession session : requestErrorList) {
				unSub(session);
			}
			try {
				// 记录请求异常的session,发送一次报错信息后,移除session队列
				List<WebSocketSession> symbolErrorList = new ArrayList<>();
				map.forEach((key, value) -> {
					// key解析
					String[] split = key.split(",");
					String symbolName = split[0];
					String lineType = split[1];
					// 币对校验
					ContractSymbolMongo mongo;
					try {
						// 币对校验
						mongo = apiContractUtils.getSymbolMongo(symbolName);
						ContractKLine kLine;
						try {
							kLine = contractUtils.selectCurrentKLine(symbolName, lineType);
						} catch (Exception e) {
							throw new OpenApiException(
									AbnormalEnum.MSG.DISCONNECTED.getCode(),
									MessageUtils.get(AbnormalEnum.MSG.DISCONNECTED.getMessage())
							);
						}
						// 类型转换
						JSONObject object = new JSONObject();
						if (kLine != null) {
							// 时间戳
							long l = kLine.getStartTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
							kLine.setTimeStamp(l);
							if (mongo.getLastPrice() != null) {
								kLine.setClosePrice(mongo.getLastPrice());
							}
							object = this.getJsonObject(kLine, lineType);
						}
						// 发送消息
						for (WebSocketSession session : value) {
							try {
								// 发送币对最新价
								session.sendMessage(OpenApiSendMessageV1.getTextMessage(
										R.ok(object)
								));
							} catch (IOException exception) {
								throw new RuntimeException(exception);
							}
						}
					} catch (OpenApiException e) {
						// 返回异常消息
						for (WebSocketSession session : value) {
							try {
								// 将捕获的异常消息发送
								session.sendMessage(OpenApiSendMessageV1.getTextMessage(
										R.failed(e.getCode(), e.getMessage())
								));
								symbolErrorList.add(session);
							} catch (IOException exception) {
								throw new RuntimeException(exception);
							}
						}
					}
				});
				// 移除请求无效的session
				for (WebSocketSession session : symbolErrorList) {
					unSub(session);
				}
			} catch (Exception e) {
				e.printStackTrace();
				log.error("webSocket---symbol_last_price---币对最优挂单Error");
			}
		},0, 500, TimeUnit.MILLISECONDS);
	}

	/**
	 * k线 参数格式重组
	 * @param kLine k线数据
	 * @param type k线类型
	 * @return jsonObject
	 */
	private JSONObject getJsonObject(ContractKLine kLine, String type) {
		JSONObject object = new JSONObject();
		object.put("e", "k_line");
		object.put("E", LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
		object.put("s", kLine.getSymbolName());
		object.put("k", new JSONObject() {{
			put("t", kLine.getStartTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
			put("T", kLine.getEndTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
			put("s", kLine.getSymbolName());
			put("i", type);
			put("f", "");
			put("L", "");
			put("o", kLine.getOpenPrice());
			put("c", kLine.getClosePrice());
			put("h", kLine.getHighPrice());
			put("l", kLine.getLowPrice());
			put("v", kLine.getDealNum());
			put("n", "");
			put("x", false);
			// 涨跌额
			BigDecimal subtract = kLine.getClosePrice().subtract(kLine.getOpenPrice());
			if (subtract.compareTo(BigDecimal.ZERO) < 0) {
				put("q", subtract.negate());
			} else {
				put("q", subtract);
			}
			// 涨跌幅
			BigDecimal divide = subtract.multiply(BigDecimal.TEN).multiply(BigDecimal.TEN).divide(kLine.getOpenPrice(),2,BigDecimal.ROUND_HALF_UP);
			put("Q", divide);
			put("V", kLine.getDealPrice());
			put("B", "");
		}});
		return object;
	}

}
